package com.atguigu.flink.sql;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Over;
import org.apache.flink.table.api.OverWindow;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.*;

/**
 * Created by Smexy on 2022/12/23
 */
public class Demo10_OverWindow
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //为了获取eventtime
        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy.<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner( (e, r) -> e.getTs());

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy)
            ;


        Table table = tableEnv.fromDataStream(ds, $("ts"),$("vc"),$("id"),$("pt").proctime(),
            $("et").rowtime());


        tableEnv.createTemporaryView("ws",table);

        //基于row的窗口
       String sql1 = " select id,ts,vc, sum(vc) over(partition by id order by et rows between unbounded preceding and current row  ) from ws   ";

       String sql2 = " select id,ts,vc, sum(vc) over(partition by id order by et rows between 2 preceding and current row  ) from ws   ";

       //基于range的窗口   eventtime  < watermark ，没有机会再去算，属于迟到的数据！
        String sql3 = " select id,ts,vc, sum(vc) over(partition by id order by et range between unbounded preceding and current row  ) from ws   ";

        String sql4 = " select id,ts,vc, sum(vc) over(partition by id order by et range between interval '2' seconds preceding and current row  ) from ws   ";

        //进行多个窗口函数的同时计算
        //TableException: Over Agg: Unsupported use of OVER windows. All aggregates must be computed on the same window.
        // 多个函数，计算的over()窗口必须一致！ 否则报错！
        String sql5 = " select id,ts,vc, " +
            "            sum(vc) over(partition by id order by et range between interval '2' seconds preceding and current row  )  sumVC ," +
            "            max(vc) over(partition by id order by et range between interval '2' seconds preceding and current row  )  maxVC " +
            "           from ws   ";


        String sql6 = " select id,ts,vc, " +
            "            sum(vc) over w  sumVC ," +
            "            max(vc) over w  maxVC " +
            "           from ws   " +
            "           window w as (partition by id order by et range between interval '2' seconds preceding and current row  )  ";



        tableEnv.sqlQuery(sql6)
               .execute()
               .print();

    }
}
